package org.nbict.iot.trash;

import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;

/**
 * Created by songseven on 18/6/11.
 */
@Deprecated
public class Topology {

    public static void main(String[] args) throws InvalidTopologyException,
            AuthorizationException, AlreadyAliveException {

        TopologyBuilder builder = new TopologyBuilder();

        KafkaSpoutConfig.Builder<String, byte[]> kafkaBuilder = new
                KafkaSpoutConfig.Builder("127.0.0.1:9092", StringDeserializer
                .class, ByteArrayDeserializer.class, "5vvokv"); //5vvokv

        kafkaBuilder.setGroupId("iot-car-group-test-12");
        KafkaSpoutConfig build = kafkaBuilder.build();
        KafkaSpout kafkaSpout = new KafkaSpout(build);

        //step1. 报文读取流
        builder.setSpout("kafkaSpout", kafkaSpout, 1);
        //step2. 报文解析流
        builder.setBolt("parserBolt", new ParserBolt(), 2)
                .shuffleGrouping("kafkaSpout");

        //step3. 校验错误流
        builder.setBolt("checksum-error", new S1Bolt(), 1)
                .shuffleGrouping("parserBolt", "S1");
        //step4. 报文头部动作
        builder.setBolt("action", new S0Bolt(), 1)
                .shuffleGrouping("parserBolt", "S0");

        //step6.
        builder.setBolt("login-logout", new S01Bolt(), 1)
                .shuffleGrouping("action", "S0-1");
        //step7. 报文信息体流
        builder.setBolt("payload", new S02Bolt(), 2)
                .shuffleGrouping("action", "S0-2");

        Config config = new Config();

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, builder
                    .createTopology());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("kafkaSpout", config, builder
                    .createTopology());
        }
    }

}
